/**
 * 
 */
package core;

import java.awt.geom.Line2D;
import java.awt.geom.Point2D;
import java.awt.geom.Rectangle2D;
import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.StringTokenizer;

import util.Constants;
import util.NBGStrParser;
import NET.sourceforge.BplusJ.BplusJ.BplusTree;

/**
 * Parinet implementation<br>
 * Read NBG data set directly Road id -> partition partition -> b+tree having
 * timestamp=>oid
 * 
 * @author Xiaohui
 * @version 0.1 rid->partition mapping is changed to spatial proximity from
 *          modulo; partition changed to 2d array
 * @version 0.2 Multiple queries allowed
 */
public class Parinet implements Constants {

	static int numPartitions = 4; // 4x4 grids
	String prefix = Constants.pathPrefix + "meta/";
	// partition table
	Partition[][] parts;
	RoadNetworkIndex rnidxer;

	/**
	 * @param args
	 * @throws Exception
	 */
	public Parinet() throws Exception {
		rnidxer = new RoadNetworkIndex();
	}

	public void buildPartitions() throws Exception {
		parts = new Partition[numPartitions][numPartitions];
		String[][] treefiles = new String[numPartitions][numPartitions];
		String[][] blockfiles = new String[numPartitions][numPartitions];
		double stepx = (oldmaxx - oldminx) / numPartitions;
		double stepy = (oldmaxy - oldminy) / numPartitions;

		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				treefiles[i][j] = prefix + "tree" + i + j;
			}
		}

		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				blockfiles[i][j] = prefix + "block" + i + j;
			}
		}
		// create partitions with the right geometry
		for (int i = 0; i < numPartitions; i++) {
			double lowx = oldminx + stepx * i;
			double highx = oldminx + stepx * (i + 1);
			if (i == numPartitions - 1) {
				highx = oldmaxx;
			}
			for (int j = 0; j < numPartitions; j++) {
				double lowy = oldminy + stepy * j;
				double highy = oldminy + stepy * (j + 1);
				if (j == numPartitions - 1) {
					highy = oldmaxy;
				}
				parts[i][j] = new Partition(treefiles[i][j], blockfiles[i][j],
						lowx, lowy, highx - lowx, highy - lowy);
			}
		}

	}

	private void printPartitions() {
		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				System.out.println(parts[i][j]);
			}
		}
	}

	String getTreeNameById(int id) {
		return prefix + "tree" + id;
	}

	String getBlockNameById(int id) {
		return prefix + "block" + id;
	}

	/**
	 * Write input lines into several files according to the geometric relations
	 * 
	 * @param rawFileName
	 * @param outFilePrefix
	 */
	static void separateIntoFiles(String rawFileName, String outFilePrefix,
			String sqlOutFile) {
		System.out.println("Separating data into partitions...");
		double stepx = (oldmaxx - oldminx) / numPartitions;
		double stepy = (oldmaxy - oldminy) / numPartitions;

		BufferedReader in;
		BufferedWriter[][] bws = new BufferedWriter[numPartitions][numPartitions];

		// separate the space and get right geometry
		Rectangle2D.Double[][] rectangles = new Rectangle2D.Double[numPartitions][numPartitions];
		for (int i = 0; i < numPartitions; i++) {
			double lowx = oldminx + stepx * i;
			double highx = oldminx + stepx * (i + 1);
			if (i == numPartitions - 1) {
				highx = oldmaxx;
			}
			for (int j = 0; j < numPartitions; j++) {
				double lowy = oldminy + stepy * j;
				double highy = oldminy + stepy * (j + 1);
				if (j == numPartitions - 1) {
					highy = oldmaxy;
				}
				rectangles[i][j] = new Rectangle2D.Double(lowx, lowy, highx
						- lowx, highy - lowy);
			}
		}

		try {
			BufferedWriter sqlOut = new BufferedWriter(new FileWriter(
					sqlOutFile));
			sqlOut.write("use traj;");
			sqlOut.newLine();
			
			// init bws, every rectangle corresponds to one writer
			for (int i = 0; i < numPartitions; i++) {
				for (int j = 0; j < numPartitions; j++) {
					String temp = outFilePrefix + i + j;
					bws[i][j] = new BufferedWriter(new FileWriter(temp));
					String table = temp.substring(temp.lastIndexOf("/") + 1);

					String sqlCreate = "DROP TABLE IF EXISTS " + table + ";\n"
							+ "CREATE TABLE  " + table + "(\n"
							+ "`state` varchar(15) DEFAULT NULL,\n"
							+ "`id` int(11) DEFAULT NULL,\n"
							+ "`seq_no` int(11) DEFAULT NULL,\n"
							+ "`class_id` int(11) DEFAULT NULL,\n"
							+ "`curr_time` int(11) DEFAULT NULL,\n"
							+ "`x_coord` decimal(13,7) DEFAULT NULL,\n"
							+ "`y_coord` decimal(13,7) DEFAULT NULL,\n"
							+ "`curr_speed` int(8) DEFAULT NULL,\n"
							+ "`roadid` int(11) unsigned DEFAULT NULL,\n"
							+ "`x_nextNode` int(10) unsigned DEFAULT NULL,\n"
							+ "`y_nextNode` int(10) unsigned DEFAULT NULL);\n";
					String sqlLoad = "LOAD DATA LOCAL INFILE '" + temp
							+ "' INTO TABLE " + table+";";
					sqlOut.write(sqlCreate);
					sqlOut.write(sqlLoad);
					sqlOut.newLine();
					sqlOut.newLine();
				}
			}
			in = new BufferedReader(new FileReader(rawFileName));
			String str;
			NBGStrParser parser;
			int count = 1;
			while ((str = in.readLine()) != null) {
				// write into right file
				if ((count % 100000) == 0)
					System.out.println("Parinet separating progress: " + count);
				count++;

				parser = new NBGStrParser(str);
				Point2D.Double pt = new Point2D.Double(parser.getCurrPosx(),
						parser.getCurrPosy());
				// check which rectangle this point belongs to, and write
				for (int i = 0; i < numPartitions; i++) {
					for (int j = 0; j < numPartitions; j++) {
						if (rectangles[i][j].contains(pt)) {
							bws[i][j].write(str);
							bws[i][j].newLine();
						}
					}

				}
			}
			in.close();
			for (int i = 0; i < bws.length; i++) {
				for (int j = 0; j < numPartitions; j++) {
					bws[i][j].close();
				}
			}
			sqlOut.close();
			System.out.println("Done!");
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * Build B-tree with data Normalized file expected.<br>
	 * 1. Distribute data into respective regions <br>
	 * 2. create b+ tree index on the regions <br>
	 * 
	 * @param normFilename
	 * @throws Exception
	 */
	public void loadData(String rawFilename) throws Exception {
		// note that disappearing data has speed ==0
		System.out
				.println("Loading data into partitions and index with B+Tree...");
		BufferedReader in;
		// x1,y1,x2,y2

		try {
			in = new BufferedReader(new FileReader(rawFilename));
			String str;
			NBGStrParser parser;
			Point2D.Double pt;
			Partition tempPartition = null;
			int count = 0;
			while ((str = in.readLine()) != null) {
				if ((count % 100000) == 0)
					System.out.println("Parinet Loading progress: " + count);
				count++;
				parser = new NBGStrParser(str);
				pt = new Point2D.Double(parser.getCurrPosx(), parser
						.getCurrPosy());
				for (int i = 0; i < numPartitions; i++) {
					for (int j = 0; j < numPartitions; j++) {
						if (parts[i][j].contains(pt)) {
							tempPartition = parts[i][j];
						}
					}
				}
				if (tempPartition != null) {
					tempPartition.oidList.add(parser.getOid());
					// update b+tree
					// timestamp=>oid
					if (tempPartition.bpt.ContainsKey(""
							+ parser.getTimepoint())) {

						String newValue = tempPartition.bpt.get(""
								+ parser.getTimepoint())
								+ "," + parser.getOid();

						tempPartition.insert(parser.getTimepoint() + "",
								newValue);
					} else {
						tempPartition.insert(parser.getTimepoint() + "", parser
								.getOid()
								+ "");
					}
				}
			}
			in.close();
			commitAll();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	/**
	 * called by load data
	 * 
	 * @param node1
	 *            of the road edge
	 * @param node2
	 *            of the road edge
	 * @return set of id of the partition intersecting with the road
	 */
	public void fillIntersectedPartids(long roadId, BplusTree btree,
			Set<Partition> qualifiedPnum) throws Exception {

		if (btree.ContainsKey("" + roadId)) {
			String nodeDataCSV = btree.get("" + roadId);
			// System.out.println("road geom: "+nodeDataCSV);
			StringTokenizer st = new StringTokenizer(nodeDataCSV, ",");
			double x11 = Double.parseDouble(st.nextToken());
			double y11 = Double.parseDouble(st.nextToken());
			double x21 = Double.parseDouble(st.nextToken());
			double y21 = Double.parseDouble(st.nextToken());
			Line2D.Double line = new Line2D.Double(x11, y11, x21, y21);
			for (int i = 0; i < numPartitions; i++) {
				// System.out.println("Partition geom: "+parts[i]);
				for (int j = 0; j < numPartitions; j++) {
					if (parts[i][j].intersectsLine(line)) {
						// System.out.println("Adding part: "+i);
						qualifiedPnum.add(parts[i][j]);
					}
				}
			}
		}
	}

	private void commitAll() throws Exception {
		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				parts[i][j].commit();
			}
		}

	}

	// get partitions by intersecting query window with road network to
	// get set of road ids
	// followed by finding the partitions that contain the road ids
	/**
	 * @param queryType
	 *            , region
	 * @return a list of road id's that are intersecting with the query region
	 * @throws Exception
	 */
	public Set<Partition> doSpatialFiltering(String queryType, String treeFile,
			QueryParser qparser) throws Exception {
		long start = System.currentTimeMillis();
		Set<Long> roadIds = rnidxer
				.RTreeQuery4Rid(qparser, treeFile, queryType);
		// System.out.println("road ids size: " + roadIds.size());
		BplusTree tree = BplusTree.ReadOnly(Constants.networkBTreeFileByEid,
				Constants.networkBlockFileByEid);
		Set<Partition> partitionNumbers = new HashSet<Partition>();
		for (long rid : roadIds) {
			// System.out.println("Now processing rid :"+rid);
			fillIntersectedPartids(rid, tree, partitionNumbers);
			// System.out.println("Now partition numbers size: "+partitionNumbers.size());
		}
		long end = System.currentTimeMillis();
		System.out.println("num partitions to examine: "
				+ partitionNumbers.size());
		System.out.println("Spatial Fileter(ms): " + (end - start));
		return partitionNumbers;
	}

	public List<QueryParser> readQueryFile(String queryFile) {
		return QueryFileReader.getParsers(queryFile);
	}

	
	/**
	 * TODO
	 * do temporal filtering in these partitions with last two numbers in // the
	 * query file
	 */
	public Set<Integer> DBTemporalFilter(List<QueryParser> queryParsers,
			Set<Partition> partitionSet) throws Exception {
		Set<Integer> objidSet = new HashSet<Integer>();
		long temporalStart = System.currentTimeMillis();
		for (QueryParser qp : queryParsers) {
			double startTime = qp.getStartTime();
			double endTime = qp.getEndTime();

			// System.out.println("start time" + startTime);
			// System.out.println("end time: " + endTime);
			for (Partition pn : partitionSet) {

				BplusTree tree = BplusTree.ReadOnly(pn.getTreeFileName(), pn
						.getBlockFileName());
				for (int i = (int) startTime; i < (int) endTime + 1; i++) {
					if (tree.ContainsKey("" + i)) {
						String val = tree.get("" + i);
						StringTokenizer st = new StringTokenizer(val, ",");
						while (st.hasMoreElements()) {
							objidSet.add(Integer.parseInt(st.nextToken()));
						}

					}
				}
			}
		}
		long temporalEnd = System.currentTimeMillis();
		System.err.println("Temporal Filter(ms): "
				+ (temporalEnd - temporalStart));
		return objidSet;
	}
	
	/**
	 * do temporal filtering in these partitions with last two numbers in // the
	 * query file
	 */
	public Set<Integer> doTemporalFilter(QueryParser qparser,
			Set<Partition> partitionSet) throws Exception {
		Set<Integer> objidSet = new HashSet<Integer>();

		double startTime = qparser.getStartTime();
		double endTime = qparser.getEndTime();
		long start = System.currentTimeMillis();
		// System.out.println("start time" + startTime);
		// System.out.println("end time: " + endTime);
		for (Partition pn : partitionSet) {
			BplusTree tree = BplusTree.ReadOnly(pn.getTreeFileName(), pn
					.getBlockFileName());
			for (int i = (int) startTime; i < (int) endTime + 1; i++) {
				if (tree.ContainsKey("" + i)) {
					String val = tree.get("" + i);
					StringTokenizer st = new StringTokenizer(val, ",");
					while (st.hasMoreElements()) {
						objidSet.add(Integer.parseInt(st.nextToken()));
					}
				}
			}
			tree.Shutdown();
			tree = null;
		}
		long end = System.currentTimeMillis();
		System.out.println("Temporal filtering(ms): " + (end - start));
		System.out.println("After temporal filtering, candidate size: "
				+ objidSet.size());
		System.out.println(objidSet);
		return objidSet;
	}

	/**
	 * Used when a second query is provided.
	 * 
	 * @throws Exception
	 */
	public void reOpenPartitions() throws Exception {
		parts = new Partition[numPartitions][numPartitions];

		String[][] treefiles = new String[numPartitions][numPartitions];
		String[][] blockfiles = new String[numPartitions][numPartitions];
		double stepx = (oldmaxx - oldminx) / numPartitions;
		double stepy = (oldmaxy - oldminy) / numPartitions;

		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				treefiles[i][j] = prefix + "tree" + i + j;
			}
		}

		for (int i = 0; i < numPartitions; i++) {
			for (int j = 0; j < numPartitions; j++) {
				blockfiles[i][j] = prefix + "block" + i + j;
			}
		}
		// reopen partitions
		for (int i = 0; i < numPartitions; i++) {
			double lowx = oldminx + stepx * i;
			double highx = oldminx + stepx * (i + 1);
			if (i == numPartitions - 1) {
				highx = oldmaxx;
			}
			for (int j = 0; j < numPartitions; j++) {
				double lowy = oldminy + stepy * j;
				double highy = oldminy + stepy * (j + 1);
				if (j == numPartitions - 1) {
					highy = oldmaxy;
				}
				parts[i][j] = Partition.reOpen(treefiles[i][j],
						blockfiles[i][j], lowx, lowy, highx - lowx, highy
								- lowy);
			}
		}
	}

	public static void main(String args[]) {
		separateIntoFiles(args[0], args[1], args[2]);
	}
}
